package com.fwmagic.flink.projectcase.streamjoin;

import org.apache.flink.api.common.functions.CoGroupFunction;
import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.tuple.Tuple5;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

/**
 * 双流join
 *
 * watermark = 一个分区(单并行的Source)/所有分区(多并行的Source)数据的最大时间 - 延迟的时间
 * 当watermart的时间大于等于窗口的结束边界，就会触发计算
 *
 * 双流join不仅要有关联字段，两个流还要在同一个时间窗口里面！！！！
 */
public class FlinkTubmblingWindowsLeftJoinDemo {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //设置时间窗口大小为10
        int windowSize =10;

        //允许数据延迟到达的时间
        long dalay = 5002L;

        //设置EventTime作为时间标准
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        //通过env设置并行度为1，即以后所有的DataStream的并行都是1
        env.setParallelism(1);

        //设置数据源
        //左流
        DataStream<Tuple3<String,String,Long>> leftSource = env.addSource(new StreamDataLeftSource());
        //leftSource.print();
        //右流
        DataStream<Tuple3<String,String,Long>> rightSource = env.addSource(new StreamDataRightSource());
       // rightSource.print();

        //设置水位线
        //("a","1",1000)
        DataStream<Tuple3<String, String, Long>> leftStream = leftSource.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<Tuple3<String, String, Long>>(Time.milliseconds(dalay)) {
            @Override
            public long extractTimestamp(Tuple3<String, String, Long> element) {
                //获取并返回数据中的EventTime
                return element.f2;
            }
        });

        //("a","hangzhou",6000)
        SingleOutputStreamOperator<Tuple3<String, String, Long>> rightStream = rightSource.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<Tuple3<String, String, Long>>(Time.milliseconds(dalay)) {
            @Override
            public long extractTimestamp(Tuple3<String, String, Long> element) {
                //获取并返回数据中的EventTime
                return element.f2;
            }
        });

        DataStream<Tuple5<String, String, String, Long, Long>> joinStream = leftStream.coGroup(rightStream)
                .where(new LeftKeySelector())
                .equalTo(new RightKeySelector())
                .window(TumblingEventTimeWindows.of(Time.seconds(windowSize)))
                .apply(new CoGroupFunction<Tuple3<String, String, Long>, Tuple3<String, String, Long>, Tuple5<String, String, String, Long, Long>>() {
                    //coGroup左流的数据和右流的数据取出来，可以key相同，并且将同一个窗口的数据取出来
                    @Override
                    public void coGroup(Iterable<Tuple3<String, String, Long>> leftElements, Iterable<Tuple3<String, String, Long>> rightElements, Collector<Tuple5<String, String, String, Long, Long>> out) throws Exception {
                        for(Tuple3<String, String, Long> leftElem :leftElements){
                            boolean hadElements =false;
                            //如果左边的流join上了右边流的rightElements就不为空
                            for(Tuple3<String, String, Long> rightElem : rightElements){
                                //将join的数据输出
                                out.collect(Tuple5.of(leftElem.f0, leftElem.f1, rightElem.f1, leftElem.f2, rightElem.f2));
                                hadElements =true;
                            }
                            if(!hadElements){
                                //没join上，给右边的数据赋为空
                                out.collect(Tuple5.of(leftElem.f0, leftElem.f1,"null", leftElem.f2, -1L));
                            }
                        }

                    }
                });
       /* //join操作
        DataStream<Tuple5<String, String, String, Long, Long>> joinStream = leftStream.join(rightStream)
                .where(new LeftKeySelector())
                .equalTo(new RightKeySelector())
                .window(TumblingEventTimeWindows.of(Time.seconds(windowSize)))
                .apply(new MyJoinFunction());
*/
        joinStream.print();

        env.execute("FlinkTubmblingWindowsInnerJoinDemo");


    }
    public static class LeftKeySelector implements KeySelector<Tuple3<String, String, Long>,String>{

        @Override
        public String getKey(Tuple3<String, String, Long> value) throws Exception {
            return value.f0;
        }
    }
    public static class RightKeySelector implements KeySelector<Tuple3<String, String, Long>,String>{

        @Override
        public String getKey(Tuple3<String, String, Long> value) throws Exception {
            return value.f0;
        }
    }


}
